#include <iterator>
#include "MarkupSTL.h"
#include "global.h"
#include "NodeServer.h"
#include "RedisClient.h"
#include "globalExtern.h"
#include <limits>



enum TIME_ID
{
	NODE_TIMERID_LOGFILE = 0x101,
	NODE_TIMERID_HEATBEAT = 0x102,
};


CGameBase* CreateGame()
{
	CNodeServer* pServer = new CNodeServer();
	return pServer;
}
CNodeServer* GetNodeServer()
{
	return (CNodeServer*)GetGameInstance();
}

CNodeServer::CNodeServer():m_pSignalEvent(NULL), m_initFlag(false)
{
}

CNodeServer::~CNodeServer()
{
	if (m_pSignalEvent)
	{
		delete m_pSignalEvent;
		m_pSignalEvent = NULL;
	}
	for (RedisClientList::iterator itor = m_redisClientList.begin(); m_redisClientList.end() != itor; ++itor)
	{
		if (itor->second)
		{
			delete (itor->second);
			itor->second = NULL;
		}
	}
	m_redisClientList.clear();

	for (std::map<std::string, std::map<int, RedisClient*> >::iterator itor = m_redisClientCopy.begin(); m_redisClientCopy.end() != itor; ++itor)
	{
		for (std::map<int, RedisClient*>::iterator itor1 = itor->second.begin(); itor->second.end() != itor1; ++itor1)
		{
			if (itor1->second)
			{
				delete (itor1->second);
				itor1->second = NULL;
			}
		}
		itor->second.clear();
	}
	m_redisClientCopy.clear();

	for (DBClientList::iterator itor = m_dbClientList.begin(); m_dbClientList.end() != itor; ++itor)
	{
		if (itor->second)
		{
			delete (itor->second);
			itor->second = NULL;
		}
	}
	m_dbClientList.clear();

	for (std::map<std::string, std::map<int, CBaseDbPart*> >::iterator itor = m_dbClientCopy.begin(); m_dbClientCopy.end() != itor; ++itor)
	{
		for (std::map<int, CBaseDbPart*>::iterator itor1 = itor->second.begin(); itor->second.end() != itor1; ++itor1)
		{
			if (itor1->second)
			{
				delete (itor1->second);
				itor1->second = NULL;
			}
		}
		itor->second.clear();
	}
	m_dbClientCopy.clear();

}

bool CNodeServer::InitGame()
{
	if (TcpListen(TGlobal::_ip.c_str(), TGlobal::_port) < 0)
	{
		return false;
	}

	m_type = SERVER_TYPE_NONE;
	m_pHost = CreateHost();
	if (NULL == m_pHost)
		return false;

	m_pHost->m_pNode = this;
	m_pHost->Init();


	const char* conf_path = TGlobal::_xmlconfig.c_str();
	if (!ReadCommonConf(conf_path))
	{
		log_boot("InitGame ReadCommonConf failed. path:[%s]", conf_path);
		return false;
	}

	if (SERVER_TYPE_MASTER != m_type && 
		SERVER_TYPE_NONE != m_type)
	{
		m_pMaster = new CClient(this);
		m_pMaster->setId(masterConf.svid);
		m_pMaster->setType(SERVER_TYPE_MASTER);
		if (0 != m_pMaster->InitConnect(masterConf.ip, masterConf.port))
		{
			log_boot("InitGame Master connect [%s:%d]  err", masterConf.ip, masterConf.port);
		}
	}
	
	unsigned int seed = (getpid() + clock());
	srand(seed);

	//this->StartTimer(NODE_TIMERID_LOGFILE, CLogic::GetCurrDayLeastTime());
	this->StartTimer(NODE_TIMERID_HEATBEAT, 60);
	return true;
}

int CNodeServer::OnBYTimeOut(int Timerid)
{
	switch (Timerid)
	{
	//case NODE_TIMERID_LOGFILE:
	//	ProcLogFileTimer();
	//	break;
	case NODE_TIMERID_HEATBEAT:
		ProcHeatBeatTimer();
		break;
	default:
		break;
	}
	return 0;
}

int CNodeServer::ProcLogFileTimer()
{
	this->StopTimer(NODE_TIMERID_LOGFILE);
	this->StartTimer(NODE_TIMERID_LOGFILE, CLogic::GetCurrDayLeastTime());

	InitDaemonLog();
	return 0;
}

int CNodeServer::ProcHeatBeatTimer()
{
	this->StartTimer(NODE_TIMERID_HEATBEAT, 60);
	
	NETOutputPacket outPkg;
	outPkg.Begin(HEART_BEATT_CMD);
	outPkg.End();

	BackTypeList::iterator iter = m_backTypeList.begin();
	for (; iter != m_backTypeList.end(); iter++)
	{
		BackList&  backList = iter->second;
		BackList::iterator iter_back = backList.begin();
		for (; iter_back != backList.end(); iter_back++)
		{
			CClient* pBack = iter_back->second;
			pBack->Send(&outPkg);
		}
	}
	return 0;
}

bool CNodeServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}


	if (markup.FindElem("log_level"))
	{
		string str = markup.GetAttrib("level");
		if (!str.empty())
		{
			int log_level = atoi(str.c_str());
			TGlobal::_log_level = log_level;
			set_log_level(log_level);
		}
	}

	if (markup.FindElem("master"))
	{
		masterConf.svid = atoi(markup.GetAttrib("svid").c_str());
		strncpy(masterConf.ip, markup.GetAttrib("ip").c_str(), sizeof(masterConf.ip) - 1);
		masterConf.port = atoi(markup.GetAttrib("port").c_str());
	}
	else if (SERVER_TYPE_MASTER != m_type)
	{
		log_error("Not Found Elem [master] in xml[%s]", strxml);
		return false;
	}

	log_debug("CNodeServer ReadCommonConf succ!!");

	return true;
}

int CNodeServer::OnClose(CClientHandler* pHandler)
{
	//if (pHandler->getId() > 0)
	{
		SpiderEvent ev;
		ev.evID = HOST_EV_CLIENT_CLOSE;
		ev.evParam.ptr = pHandler;
		m_pHost->ProcEvent(HOST_EV_CLIENT_CLOSE, &ev);
		this->ProcServerClose(pHandler);
	}
		
	return 0;
}

int CNodeServer::ProcessPacket(CClientHandler * pHandler, NETInputPacket* inputPkg)
{
	log_debug("cmd=0x%x, %d ", inputPkg->GetCmdType(), inputPkg->GetCmdType());

	int cmd = inputPkg->GetCmdType();

	switch (cmd)
	{
	case SERVER_TYPE_REGISTER_SRV_DATA:
		this->ProcServerRegister(inputPkg, pHandler);
		break;
	case  MASTER_REGISTER_SRV_DATA:
		this->ProcServerRegister(inputPkg, pHandler);
		break;
	case ONLINE_STATE_QUERY_RSP_SYS:
		ProcOnlineQueryRsp(inputPkg, pHandler);
		break;
	case SET_LOG_LEVEL: // 走phpagent直连过来的
		ProcSetLogLevel(inputPkg, pHandler);
		break;
	default:
		m_pHost->ProcPacket(inputPkg, pHandler);
		break;
	}

	return 0;
}

int CNodeServer::OnBackClose(ClientBase* pBackClient)
{
	CClient* pCClient = (CClient*)pBackClient;
	
	if (pCClient->getType() > 0)
	{
		SpiderEvent ev;
		ev.evID = HOST_EV_BACK_CLOSE;
		ev.evParam.ptr = pCClient;
		m_pHost->ProcEvent(HOST_EV_BACK_CLOSE, &ev);

		pCClient->StartReconnectTimer(2);
		return -1;
	}

	return 0;
}

int CNodeServer::OnBackConnected(ClientBase* pBackClient)
{
	CClient* pCClient = (CClient*)pBackClient;
	if (SERVER_TYPE_MASTER == pCClient->getType())
	{
		this->ReqCacheConf(pBackClient);
	}
	else
	{//
		this->RegisterToServer(pBackClient);
	}

	SpiderEvent ev;
	ev.evID = HOST_EV_BACK_CONNECTED;
	ev.evParam.ptr = pCClient;
	m_pHost->ProcEvent(HOST_EV_BACK_CONNECTED, &ev);


	if (SERVER_TYPE_DISPATCH == pCClient->getType())
	{
		for (int i = 0; i < m_subscribeCmds.size(); i++)
		{
			NETOutputPacket outPkg; 
			outPkg.Begin(DISPATCH_SRV_REGISTER_CMD);
			outPkg.WriteInt(m_subscribeCmds[i]);
			outPkg.End(); 
			pCClient->Send(&outPkg);
		}
	}

	return 0;
}

int CNodeServer::ProccessBackPacket(ClientBase* pBackClient, NETInputPacket* inputPkg)
{
	log_debug("cmd=0x%x,  ", inputPkg->GetCmdType());

	int cmd = inputPkg->GetCmdType();

	switch (cmd)
	{
	case MASTER_QUERY_CACHE_CONF:
		ProcRecvCacheConf(pBackClient, inputPkg);
		break;
	case MASTER_PUSH_NEW_CONN:
		ProcPushNewConn(inputPkg, pBackClient);
		break;
	case SERVICEMANAGER_SRV_PUBLISH_SERVICE:
		ProcPublishService(inputPkg, pBackClient);
		break;
	case ONLINE_STATE_QUERY_RSP_SYS:
		ProcOnlineQueryRsp(inputPkg, pBackClient);
		break;
	case SET_LOG_LEVEL: // 走dispatch过来的
		ProcSetLogLevel(inputPkg, pBackClient);
		break;
	default:
		m_pHost->ProcPacket(inputPkg, pBackClient);
		break;
	}

	return 0;
}

int CNodeServer::ProcPushNewConn(NETInputPacket* inputPkg, ClientBase* pBackClient)
{
	ServerData connItem;
	connItem.type = inputPkg->ReadInt();
	connItem.svid = inputPkg->ReadInt();
	connItem.ip = inputPkg->ReadString();
	connItem.port = inputPkg->ReadInt();
	connItem.param1 = inputPkg->ReadInt();
	connItem.param2 = inputPkg->ReadInt();
	connItem.param3 = inputPkg->ReadInt();
	connItem.param_str = inputPkg->ReadString();

	SpiderEvent ev;
	ev.evID = HOST_EV_NEW_CONN_CHECK;
	ev.evParam.ptr = &connItem;
	int ret = m_pHost->ProcEvent(HOST_EV_NEW_CONN_CHECK, &ev);
	if (ret < 0)
	{
		return 0;
	}

	log_error(" type=%d, svid=%d, ip:[%s:%d],  parm1=%d,",
		connItem.type, connItem.svid, connItem.ip.c_str(), connItem.port, connItem.param1);


	CClient* pBack = NULL;
	BackList&  backList = m_backTypeList[connItem.type];

	BackList::iterator iter = backList.find(connItem.svid);
	if (iter == backList.end())
	{
		pBack = new CClient(this);
		pBack->setType(connItem.type);
		pBack->setId(connItem.svid);
		backList[connItem.svid] = pBack;

		if (0 != pBack->InitConnect(connItem.ip, connItem.port))
		{
			log_error("InitGame New type=%d, connect [%s:%d]  err", connItem.type, connItem.ip.c_str(), connItem.port);
		}
	}
	else
	{
		pBack = iter->second;
		if (pBack->GetPort() != connItem.port ||
			pBack->GetSIP() != connItem.ip)
		{// svid change ip or port
			backList.erase(iter);
			pBack->setType(0);
			pBack->setId(0);

			pBack = new CClient(this);
			pBack->setType(connItem.type);
			pBack->setId(connItem.svid);
			backList[connItem.svid] = pBack;

			if (0 != pBack->InitConnect(connItem.ip, connItem.port))
			{
				log_error("InitGame New type=%d,  connect [%s:%d]  err", connItem.type, connItem.ip.c_str(), connItem.port);
			}
		}

		// or  the same ip and port, no change
	}

	return 0;
}


int CNodeServer::ProcPublishService(NETInputPacket* inputPkg, ClientBase* pBackClient)
{
	int server_type = inputPkg->ReadInt();
	int alloc_type = inputPkg->ReadInt();
	int switch_time = inputPkg->ReadInt();
	int active_count = inputPkg->ReadInt();
	std::vector<int>  svidVect;
	for (int i = 0; i < active_count; i++)
	{
		svidVect.push_back(inputPkg->ReadInt());
	}

	ServiceConf& serviceConf = m_serviceConfList[server_type];
	serviceConf.server_type = server_type;
	serviceConf.alloc_type = alloc_type;
	serviceConf.switch_time = switch_time;
	serviceConf.active_count = active_count;

	struct timeval tv;
	gettimeofday(&tv, NULL);
	int64 tt = (tv.tv_sec * 1000 + tv.tv_usec / 1000);

	ServiceVect& serviceVect = m_serviceTypeList[server_type];
	if (serviceVect.size() != svidVect.size())
	{ 
		serviceVect.clear();
		for (int i = 0; i < svidVect.size(); i++)
		{
			ServiceItem item;
			item.svid = svidVect[i];
			item.isOK = false;
			item.start_time = tt + switch_time;
			serviceVect.push_back(item);
		}
	}
	else
	{
		for (int i = 0; i < svidVect.size(); i++)
		{
			if (serviceVect[i].svid != svidVect[i])
			{
				serviceVect[i].svid = svidVect[i];
				serviceVect[i].isOK = false;
				serviceVect[i].start_time = tt + switch_time;
			}
		}
	}
	SpiderEvent ev;
	ev.evID = HOST_EV_PUBULISH_SERVICE;
	ev.evParam.ptr = &m_serviceConfList[server_type];
	m_pHost->ProcEvent(HOST_EV_PUBULISH_SERVICE, &ev);

#if 1  //test code
	std::string str_log;
	char str_buf[64] = { 0 };
	snprintf(str_buf, 63, " provider_type=%d, active_count=%d, alloc_type=%d,  switch_time=%d, ", 
		server_type, active_count, alloc_type, switch_time);
	str_log += str_buf;
	for (int i = 0; i < serviceVect.size(); i++)
	{
		snprintf(str_buf, 63, " [idx=%d,svid=%d,isOk=%d],", i, serviceVect[i].svid, serviceVect[i].isOK);
		str_log += str_buf;
	}
	
	log_error(" %s ", str_log.c_str());
#endif

	return 0;
}

int CNodeServer::ProcOnlineQueryRsp(NETInputPacket* inputPkg, CCSocketHandler* handler)
{
	unsigned int uid = inputPkg->ReadInt();
	unsigned int agentId = inputPkg->ReadInt();
	char inner[10240];
	int len = inputPkg->ReadBinary(inner, sizeof(inner));
	if (len > 0 && agentId > 0)
	{
		NETOutputPacket outPkg;
		outPkg.Copy(inner, len);

		//SendPacketFromBack(&outPkg, SERVER_TYPE_AGENT, agentId);
		SendPacketToSrv(&outPkg, SERVER_TYPE_AGENT, agentId);
	}
	else {
		NETInputPacket input;
		input.Copy(inner, len);
		input.SetSrcSvid(uid);
		SpiderEvent ev;
		ev.evID = HOST_EV_CLIENT_OFFLINE;
		ev.evParam.ptr = &input;
		m_pHost->ProcEvent(HOST_EV_CLIENT_OFFLINE, &ev);
	}

	log_debug("uid,%d, agent,%d, len,%d, inner[0],%d", uid, agentId, len, inner[8]);
	return 0;
}

int CNodeServer::ProcSetLogLevel(NETInputPacket* inputPkg, CCSocketHandler* handler)
{
	int log_level = inputPkg->ReadInt();
	TGlobal::_log_level = log_level;
	set_log_level(log_level);
	// todo: 重写配置文件xml中的日志级别
	return 0;
}

int CNodeServer::ProcEvent(int ev_id, SpiderEvent* pEvent)
{
	if (NODE_EV_SET_TYPE == ev_id)
	{
		this->m_type = pEvent->evParam.iParam;
	}
	else if (NODE_EV_SET_CACHE == ev_id)
	{
		vector<CONFITEM>* pItem = (vector<CONFITEM>*)pEvent->evParam.ptr;
		this->conf_item = *pItem;
	}
	else if (NODE_EV_SET_SVR_REG_PARAM == ev_id)
	{
		SvrRegParam* pReg = (SvrRegParam*)pEvent->evParam.ptr;
		this->svrRegParam = *pReg;
	}
	else if (NODE_EV_SET_SUBSCRIBE_CMD == ev_id)
	{
		std::vector<int>* pCmds = (std::vector<int>*)pEvent->evParam.ptr;
		m_subscribeCmds  = *pCmds;
	}
	return 0;
}

int CNodeServer::SendPacketToService(CCPacketBase* pPkg, int type, int mod_num)
{

	log_debug("  server_type=%d, mod_num=%d, ", type, mod_num );
	SendPkgQueue();
	ServiceConfMap::iterator  iter_ = m_serviceConfList.find(type);
	if (iter_ == m_serviceConfList.end())
	{
		this->AddServicePkgList(pPkg, type, mod_num);
		return 0;
	}
	ServiceConf& serviceConf = iter_->second;
	ServiceVect& serviceVect = m_serviceTypeList[type];
	if (serviceVect.size() <= 0)
	{
		this->AddServicePkgList(pPkg, type, mod_num);
		return 0;
	}

	int idx = 0;
	if (ALLOC_MOD == serviceConf.alloc_type)
	{
		idx = mod_num%serviceVect.size();
	}
	else
	{
		idx = random() % serviceVect.size();
	}
	int svid = serviceVect[idx].svid;
	if (serviceVect[idx].isOK)
	{
		this->SendPacketToSrv(pPkg, type, svid);
	}
	else
	{
		struct timeval tv;
		gettimeofday(&tv, NULL);
		int64 tt = (tv.tv_sec * 1000 + tv.tv_usec / 1000);
		if (tt > serviceVect[idx].start_time)
		{
			serviceVect[idx].isOK = true;
			this->SendPacketToSrv(pPkg, type, svid);
		}
		else
		{// or add into send queue
			this->AddServicePkgList(pPkg, type, mod_num);
		}
	}

#if 1  //test code
	std::string str_log;
	char str_buf[64] = { 0 };
	snprintf(str_buf, 63, " provider_type=%d, active_count=%d, alloc_type=%d, ",
		serviceConf.server_type, serviceConf.active_count, serviceConf.alloc_type);
	str_log += str_buf;
	for (int i = 0; i < serviceVect.size(); i++)
	{
		snprintf(str_buf, 63, " [idx=%d,svid=%d,isOk=%d],", i, serviceVect[i].svid, serviceVect[i].isOK);
		str_log += str_buf;
	}

	log_debug(" %s ", str_log.c_str());
#endif

	return 0;
}

int CNodeServer::SendtoSrv(CCPacketBase* pPkg, const unsigned short dsttype, const unsigned int mod_uid)
{
	static unsigned int s_randnum = 0;
	ServiceConfMap::iterator iter = m_serviceConfList.find(dsttype);
	if (m_serviceConfList.end() != iter)
	{
		ServiceConf& serviceConf = iter->second;
		ServiceVect& serviceVect = m_serviceTypeList[dsttype];
		if (serviceVect.size() == 0)
		{
			return -1;
		}

		int idx = 0;
		if (ALLOC_MOD == serviceConf.alloc_type)
		{
			idx = mod_uid % serviceVect.size();
		}
		else
		{
			s_randnum = ++s_randnum % numeric_limits<unsigned int>::max();
			idx = s_randnum % serviceVect.size();
		}

		unsigned int svid = serviceVect[idx].svid;
		if (serviceVect[idx].isOK)
		{
			SendPacketToSrv(pPkg, dsttype, svid);
		}
		else
		{
			struct timeval tv;
			gettimeofday(&tv, NULL);
			int64 tt = (tv.tv_sec * 1000 + tv.tv_usec / 1000);
			if (tt > serviceVect[idx].start_time)
			{
				serviceVect[idx].isOK = true;
				SendPacketToSrv(pPkg, dsttype, svid);
			}
			else
			{
				return -1;
			}
		}
	}
	else
	{
		unsigned int mod_n = mod_uid;
		if (0 == mod_n)
		{
			s_randnum = ++s_randnum % numeric_limits<unsigned int>::max();
			mod_n = s_randnum;
		}

		CCSocketHandler* handler = NULL;
		BackTypeList::iterator itor = m_backTypeList.find(dsttype);
		if (m_backTypeList.end() != itor)
		{
			if (itor->second.size() > 0)
			{
				int index = mod_n % itor->second.size();
				BackList::iterator itor1 = itor->second.begin();
				for (int i = 0; itor->second.end() != itor1 && i <= index; ++itor1, ++i)
				{
					if (index == i)
					{
						handler = itor1->second;
						break;
					}
				}
			}
		}
		else
		{
			SrvTypeList::iterator itor2 = m_svrTypeList.find(dsttype);
			if (m_svrTypeList.end() != itor2)
			{
				if (itor2->second.size() > 0)
				{
					int index = mod_n % itor2->second.size();
					SrvDataList::iterator itor3 = itor2->second.begin();
					for (int i = 0; itor2->second.end() != itor3 && i <= index; ++itor3, ++i)
					{
						if (index == i)
						{
							handler = itor3->second->pHandler;
							break;
						}
					}
				}
			}
		}
		if (!handler)
			return -1;

		handler->Send(pPkg);
	}

	return 0;
}

int CNodeServer::SendPacket(CCPacketBase* pPkg, const unsigned short dsttype, const unsigned int dstid)
{
	SendPkgQueue();
	return this->SendPacketToSrv(pPkg, dsttype, dstid);
}


int  CNodeServer::SendPacketToSrv(CCPacketBase* pPkg, const unsigned short dsttype, const unsigned int dstid)
{
	if (this->SendPacketFromBack(pPkg, dsttype, dstid))
	{
		return 0;
	}
	else if (this->SendPacketFromClient(pPkg, dsttype, dstid))
	{
		return 0;
	}
	else if (this->SendPacketFromDispatch(pPkg, dsttype, dstid))
	{
		return 0;
	}
	return -1;
}

bool CNodeServer::SendPacketFromBack(CCPacketBase* pPkg, const unsigned short dsttype, const unsigned int dstid)
{
	BackList&  backList = m_backTypeList[dsttype];
	BackList::iterator iter = backList.end();
	if (dstid > 0 && backList.size() > 0)
	{
		iter = backList.find(dstid);
	}
	else if (backList.size() > 0)
	{
		iter = backList.begin();
	}
	if (iter != backList.end())
	{
		CClient* pBack = iter->second;
		pBack->Send(pPkg);
	}
	else
	{
		return false;
	}
	return true;
}

bool CNodeServer::SendPacketFromClient(CCPacketBase* pPkg, const unsigned short dsttype, const unsigned int dstid)
{
	SrvDataList& serverList = m_svrTypeList[dsttype];
	SrvDataList::iterator iter = serverList.end(); 
	if (dstid > 0 && serverList.size() > 0)
	{
		iter = serverList.find(dstid);
	}
	else if (serverList.size() > 0)
	{
		iter = serverList.begin();
	}

	if (iter != serverList.end())
	{
		CClientHandler* pClient= iter->second->pHandler;
		pClient->Send(pPkg);
	}
	else
	{
		return false;
	}
	return true;
}

bool CNodeServer::SendPacketFromDispatch(CCPacketBase* pPkg, const unsigned short dsttype, const unsigned int dstid)
{
	return false;
}


CacheConf* CNodeServer::CreateCacheConf(std::string name)
{
	CacheConf* pConf = NULL;
	CacheConfList::iterator iter = m_cacheConfList.find(name);
	if (iter == m_cacheConfList.end())
	{
		pConf = new CacheConf();
		pConf->reset();
		m_cacheConfList[name] = pConf;
	}
	else
	{
		pConf = iter->second;
	}
	return pConf;
}

CacheConf* CNodeServer::GetCahceConf(std::string name)
{
	CacheConf* pConf = NULL;
	CacheConfList::iterator iter = m_cacheConfList.find(name);
	if (iter != m_cacheConfList.end())
	{
		pConf = iter->second;
	}
	return pConf;
}

DBConf* CNodeServer::CreateDBConf(std::string name)
{
	DBConf* pConf = NULL;
	DBConfList::iterator iter = m_dbConfList.find(name);
	if (iter == m_dbConfList.end())
	{
		pConf = new DBConf();
		pConf->reset();
		m_dbConfList[name] = pConf;
	}
	else
	{
		pConf = iter->second;
	}
	return pConf;
}

DBConf* CNodeServer::GetDBConf(std::string name)
{
	DBConf* pConf = NULL;
	DBConfList::iterator iter = m_dbConfList.find(name);
	if (iter != m_dbConfList.end())
	{
		pConf = iter->second;
	}
	return pConf;
}

RedisClient* CNodeServer::CreateCacheClinet(const std::string& name, const int index)
{
	RedisClient* pRedis = NULL;
	if (0 == index)
	{
		RedisClientList::iterator iter = m_redisClientList.find(name);
		if (iter == m_redisClientList.end())
		{
			pRedis = new RedisClient();
			m_redisClientList[name] = pRedis;
		}
		else
		{
			pRedis = iter->second;
		}
	}
	else if (index > 0)
	{
		std::map<std::string, std::map<int, RedisClient*> >::iterator iter = m_redisClientCopy.find(name);
		if (m_redisClientCopy.end() == iter)
		{
			pRedis = new RedisClient();
			m_redisClientCopy[name].insert(std::make_pair(index, pRedis));
		}
		else
		{
			std::map<int, RedisClient*>::const_iterator itor1 = iter->second.find(index);
			if (iter->second.end() == itor1)
			{
				pRedis = new RedisClient();
				iter->second.insert(std::make_pair(index, pRedis));
			}
			else
			{
				pRedis = itor1->second;
			}
		}
	}

	return pRedis;
}

RedisClient* CNodeServer::GetCacheClient(const std::string& name, const int index)
{
	RedisClient* pRedis = NULL;
	if (0 == index)
	{
		RedisClientList::iterator iter = m_redisClientList.find(name);
		if (iter != m_redisClientList.end())
		{
			pRedis = iter->second;
		}
	}
	else if (index > 0)
	{
		std::map<std::string, std::map<int, RedisClient*> >::const_iterator iter = m_redisClientCopy.find(name);
		if (m_redisClientCopy.end() != iter)
		{
			std::map<int, RedisClient*>::const_iterator itor1 = iter->second.find(index);
			if (iter->second.end() != itor1)
			{
				pRedis = itor1->second;
			}
		}
	}

	return pRedis;
}


CBaseDbPart* CNodeServer::CreateDBHandler(const std::string& name, const int index)
{
	CBaseDbPart* pHandler = NULL;
	if (0 == index)
	{
		DBClientList::iterator iter = m_dbClientList.find(name);
		if (iter == m_dbClientList.end())
		{
			pHandler = new CBaseDbPart();
			m_dbClientList[name] = pHandler;
		}
		else
		{
			pHandler = iter->second;
		}
	}
	else if (index > 0)
	{
		std::map<std::string, std::map<int, CBaseDbPart*> >::iterator iter = m_dbClientCopy.find(name);
		if (m_dbClientCopy.end() == iter)
		{
			pHandler = new CBaseDbPart();
			m_dbClientCopy[name].insert(std::make_pair(index, pHandler));
		}
		else
		{
			std::map<int, CBaseDbPart*>::const_iterator itor1 = iter->second.find(index);
			if (iter->second.end() == itor1)
			{
				pHandler = new CBaseDbPart();
				iter->second.insert(std::make_pair(index, pHandler));
			}
			else
			{
				pHandler = itor1->second;
			}
		}	
	}

	return pHandler;
}

CBaseDbPart* CNodeServer::GetDBHandler(const std::string& name, const int index)
{
	CBaseDbPart* pHandler = NULL;
	if (0 == index)
	{
		DBClientList::iterator iter = m_dbClientList.find(name);
		if (iter != m_dbClientList.end())
		{
			pHandler = iter->second;
		}
	}
	else if (index > 0)
	{
		std::map<std::string, std::map<int, CBaseDbPart*> >::const_iterator iter = m_dbClientCopy.find(name);
		if (m_dbClientCopy.end() != iter)
		{
			std::map<int, CBaseDbPart*>::const_iterator itor1 = iter->second.find(index);
			if (iter->second.end() != itor1)
			{
				pHandler = itor1->second;
			}
		}
	}
	return pHandler;
}

int CNodeServer::ProcRecvCacheConf(ClientBase* pBackClient, NETInputPacket* inputPkg)
{

	int count = inputPkg->ReadInt();
	for (int i = 0; i < count; i++)
	{
		string cache_name = inputPkg->ReadString();
		BYTE type = inputPkg->ReadByte();
		BYTE ret = inputPkg->ReadByte();
		if (ret  > 0)
		{
			if (MASTER_CONF_REDIS == type)
			{
				CacheConf* pConf = this->CreateCacheConf(cache_name);

				pConf->name = cache_name;
				pConf->ip = inputPkg->ReadString();
				pConf->port = inputPkg->ReadInt();
				log_error("cache_name=%s,  ip=[%s:%d]", cache_name.c_str(), pConf->ip.c_str(), pConf->port);

			}
			else if (MASTER_CONF_DB == type)
			{
				DBConf* pConf = this->CreateDBConf(cache_name);

				pConf->name = cache_name;
				pConf->db_name = inputPkg->ReadString();
				pConf->ip = inputPkg->ReadString();
				pConf->port = inputPkg->ReadInt();
				pConf->user = inputPkg->ReadString();
				pConf->passwd = inputPkg->ReadString();

				log_error("name=%s,   db_name=%s, ip=[%s:%d], user=%s, ",
					cache_name.c_str(), pConf->db_name.c_str(), pConf->ip.c_str(), pConf->port, pConf->user.c_str());

			}
		}
		else
		{
			log_error("cache_name=%s, type=%d, err no conf find !!!", cache_name.c_str(), type);
		}
	}

	if (this->ConnectToCache())
	{
		this->RegisterToMaster(pBackClient);

		if (!m_initFlag)
		{
			m_pHost->ProcEvent(HOST_EV_GAME_INIT, NULL);
			m_initFlag = true;
		}
		
		log_error("init server success  ok !!!");
	}
	else
	{
		log_error("init server  conn cache/db  errr!!!");
		exit(1);
	}
	
	return 0;
}

bool CNodeServer::ConnectToCache()
{

	for (int i = 0; i < conf_item.size(); i++)
	{
		if (MASTER_CONF_REDIS == conf_item[i].type)
		{
			CacheConf* pConf = GetCahceConf(conf_item[i].name);
			if (NULL == pConf)
			{
				log_error("cache_name=%s, type=%d, err no conf find", conf_item[i].name.c_str(), conf_item[i].type);
				return false;
			}
			if (0 == conf_item[i].count)
				conf_item[i].count = 1;

			bool connflag = false;
			for (unsigned int j = 0; j < conf_item[i].count; ++j)
			{
				RedisClient* pRedis = CreateCacheClinet(pConf->name, j);
				bool bReconn = false;
				if (pRedis->IsValid())
				{
					string ip1_ = pRedis->m_strHost;
					string ip2_ = pConf->ip;
					if ((pRedis->m_nPort != pConf->port) || (ip1_ != ip2_))
					{
						bReconn = true;
					}
				}
				else
				{
					bReconn = true;
				}

				if (bReconn)
				{
					if (-1 == pRedis->InitRedisConnect(pConf->ip.c_str(), pConf->port))
					{
						log_error("connect to cache=%s, [%s,%d] error",
							pConf->name.c_str(), pConf->ip.c_str(), pConf->port);
						return false;
					}
					connflag = true;

					log_error("connect cache=%s, [%s,%d] success",
						pConf->name.c_str(), pConf->ip.c_str(), pConf->port);

				}
			}

			if (connflag)
			{
				SpiderEvent ev;
				ev.evID = HOST_EV_CACHE_CONN_SUCCESS;
				ev.evParam.ptr = &conf_item[i];
				m_pHost->ProcEvent(HOST_EV_CACHE_CONN_SUCCESS, &ev);
			}
		}
		else if (MASTER_CONF_DB == conf_item[i].type)
		{
			DBConf* pConf = GetDBConf(conf_item[i].name);
			if (NULL == pConf)
			{
				log_error("db_name=%s, type=%d, err no conf find", conf_item[i].name.c_str(), conf_item[i].type);
				return false;
			}
			if (0 == conf_item[i].count)
				conf_item[i].count = 1;

			bool connflag = false;
			for (unsigned int j = 0; j < conf_item[i].count; ++j)
			{
				CBaseDbPart* pDBHandler = CreateDBHandler(conf_item[i].name, j);
				if (NULL == pDBHandler->m_pMysqlConn)
				{
					TDatabase_Param dbParam;
					dbParam.host = pConf->ip;
					dbParam.port = pConf->port;
					dbParam.user = pConf->user;
					dbParam.password = pConf->passwd;
					dbParam.db = pConf->db_name;

					if (!pDBHandler->ConnectDB(dbParam))
					{//�������ݿ�
						log_error("ConnectDB [%s:%d]  db_name=%s,  user=%s, failed\n",
							pConf->ip.c_str(), pConf->port, pConf->db_name.c_str(), pConf->user.c_str());
						return false;
					}
					log_error("ConnectDB [%s:%d]  db_name=%s,  user=%s, success\n",
						pConf->ip.c_str(), pConf->port, pConf->db_name.c_str(), pConf->user.c_str());

					connflag = true;
				}
			}

			if (connflag)
			{
				SpiderEvent ev;
				ev.evID = HOST_EV_CACHE_CONN_SUCCESS;
				ev.evParam.ptr = &conf_item[i];
				m_pHost->ProcEvent(HOST_EV_CACHE_CONN_SUCCESS, &ev);
			}
		}
	}

	return true;
}


int CNodeServer::ReqCacheConf(ClientBase* pBackClient)
{
	NETOutputPacket outPkg;
	outPkg.Begin(MASTER_QUERY_CACHE_CONF);
	outPkg.WriteInt(conf_item.size());
	for (size_t i = 0; i < conf_item.size(); i++)
	{
		outPkg.WriteByte(conf_item[i].type);
		outPkg.WriteString(conf_item[i].name);
	}
	outPkg.End();

	pBackClient->Send(&outPkg);
	return 0;
}

int CNodeServer::RegisterToServer(ClientBase* pBackClient)
{
	NETOutputPacket outPkg;
	outPkg.Begin(SERVER_TYPE_REGISTER_SRV_DATA);
	outPkg.WriteInt(this->m_type);
	outPkg.WriteInt(TGlobal::_svid);
	outPkg.WriteString(TGlobal::_ip);
	outPkg.WriteInt(TGlobal::_port);

	outPkg.WriteInt(this->svrRegParam.param1);
	outPkg.WriteInt(this->svrRegParam.param2);
	outPkg.WriteInt(this->svrRegParam.param3);
	outPkg.WriteString(this->svrRegParam.str_param);
	outPkg.End();
	pBackClient->Send(&outPkg);

	return 0;
}


int CNodeServer::RegisterToMaster(ClientBase* pBackClient)
{
	NETOutputPacket opk;
	opk.Begin(MASTER_REGISTER_SRV_DATA);
	opk.WriteInt(this->m_type);
	opk.WriteInt(TGlobal::_svid);
	opk.WriteString(TGlobal::_ip);
	opk.WriteInt(TGlobal::_port);

	opk.WriteInt(this->svrRegParam.param1);  
	opk.WriteInt(this->svrRegParam.param2);
	opk.WriteInt(this->svrRegParam.param3);
	opk.WriteString(this->svrRegParam.str_param);
	opk.End();
	pBackClient->Send(&opk);

	return 0;
}

int CNodeServer::ProcServerRegister(NETInputPacket* inputPkg, CClientHandler* pHandler)
{
	int type = inputPkg->ReadInt();
	int svid = inputPkg->ReadInt();
	std::string ip = inputPkg->ReadString();
	int port = inputPkg->ReadInt();
	int param1 = inputPkg->ReadInt();
	int param2 = inputPkg->ReadInt();
	int param3 = inputPkg->ReadInt();
	std::string param_str = inputPkg->ReadString();
	
	ServerData* pItem = NULL;

	SrvDataList& serverList = m_svrTypeList[type];
	SrvDataList::iterator iter = serverList.find(svid);

	if (iter != serverList.end())
	{
		pItem = iter->second;
		if (pItem->pHandler != pHandler)
		{
			pItem->pHandler->setType(0);
			pItem->pHandler->setId(0);
		}
	}
	else
	{
		pItem = new ServerData();
		pItem->type = type;
		pItem->svid = svid;
		pItem->ip = ip;
		pItem->port = port;
		pItem->param1 = param1;
		pItem->param2 = param2;
		pItem->param3 = param3;
		pItem->param_str = param_str;	
	}

	pHandler->setType(type);
	pHandler->setId(svid);
	pItem->pHandler = pHandler;

	serverList[svid] = pItem;
	log_debug(" server_type=%d, svid=%d, ip:[%s:%d], param1=%d,  Register, ",
		type, svid, ip.c_str(), port, param1);

	SpiderEvent ev;
	ev.evID = HOST_EV_CLIENT_REGISTER;
	ev.evParam.ptr = pItem;
	m_pHost->ProcEvent(HOST_EV_CLIENT_REGISTER, &ev);

	return 0;
}

int CNodeServer::ProcServerClose(CClientHandler* pHandler)
{
	int type = pHandler->getType();
	int svid = pHandler->getId();
	
	ServerData* pItem = NULL;
	SrvDataList& serverList = m_svrTypeList[type];
	SrvDataList::iterator iter = serverList.find(svid);

	if (iter != serverList.end())
	{
		pItem = iter->second;
		log_debug(" server_type=%d, svid=%d, ip:[%s:%d], param1=%d,  close conn, ",
			type, svid, pItem->ip.c_str(), pItem->port, pItem->param1);
		
		serverList.erase(svid);
		delete pItem;
	}

	pHandler->setType(0);
	pHandler->setId(0);
	return 0;
}


int CNodeServer::AddServicePkgList(CCPacketBase* pPkg, int type, int mod_num)
{
	struct timeval tv;
	gettimeofday(&tv, NULL);
	int tt = (int)tv.tv_sec ;

	PacketBuffer* pBuff = new PacketBuffer();
	pBuff->dst_type = type;
	pBuff->dst_svid = 0;
	pBuff->mod_num = mod_num;
	pBuff->addedTime = tt;
	pBuff->cache.append(pPkg->packet_buf(), pPkg->packet_size());
	m_servicePkgList.push_back(pBuff);


	if (m_servicePkgList.size() > 999)
	{// limit the packet count
		int del_count = m_servicePkgList.size() - 999;
		for (int i = 0; i < del_count; i++)
		{
			m_servicePkgList.pop_front();
		}
	}
}

int CNodeServer::AddSrvPkgList(CCPacketBase* pPkg, int type, int svid)
{
	struct timeval tv;
	gettimeofday(&tv, NULL);
	int tt = (int)tv.tv_sec;

	PacketBuffer* pBuff = new PacketBuffer();
	pBuff->dst_type = type;
	pBuff->dst_svid = svid;
	pBuff->mod_num = 0;
	pBuff->addedTime = tt;
	pBuff->cache.append(pPkg->packet_buf(), pPkg->packet_size());
	m_srvPkgList.push_back(pBuff);


	if (m_srvPkgList.size() > 999)
	{// limit the packet count
		int del_count = m_srvPkgList.size() - 999;
		for (int i = 0; i < del_count; i++)
		{
			m_srvPkgList.pop_front();
		}
	}
}


int CNodeServer::SendPkgQueue()
{
	if (m_servicePkgList.size() <= 0)
		return 0;

	struct timeval tv;
	gettimeofday(&tv, NULL);
	int currTime = (int)tv.tv_sec;
	int node_count = m_servicePkgList.size();
	for (int i = 0; i < node_count; i++)
	{
		if (m_servicePkgList.size() <= 0)
			return 0;
		
		PacketBuffer*  pBuff = m_servicePkgList.front();
		if ((currTime - pBuff->addedTime) > 5)
		{// wait more than 5s,  don't send 
			m_servicePkgList.pop_front();
			delete pBuff;		
		}
		else
		{//
			ServiceConfMap::iterator  iter_ = m_serviceConfList.find(pBuff->dst_type);
			if (iter_ == m_serviceConfList.end())
				return 0;
			
			ServiceConf& serviceConf = iter_->second;
			ServiceVect& serviceVect = m_serviceTypeList[pBuff->dst_type];
			if (serviceVect.size() <= 0)
				return 0;
	
			int mod_num = pBuff->mod_num;
			int idx = 0;
			if (ALLOC_MOD == serviceConf.alloc_type)
			{
				idx = mod_num % serviceVect.size();
			}
			else
			{
				idx = random() % serviceVect.size();
			}
			int svid = serviceVect[idx].svid;
			if (serviceVect[idx].isOK)
			{
				NETOutputPacket outPkg;
				outPkg.Copy(pBuff->cache.data(), pBuff->cache.data_len());
				this->SendPacketToSrv(&outPkg, pBuff->dst_type, svid);
			}
			else
			{
				struct timeval tv;
				gettimeofday(&tv, NULL);
				int64 tt = (tv.tv_sec * 1000 + tv.tv_usec / 1000);
				if (tt > serviceVect[idx].start_time)
				{
					NETOutputPacket outPkg;
					outPkg.Copy(pBuff->cache.data(), pBuff->cache.data_len());

					serviceVect[idx].isOK = true;
					this->SendPacketToSrv(&outPkg, pBuff->dst_type, svid);
				}
				else 
				{// wait switch time, the front node send err
					return 0;
				}
			}

#if 1  //test code
			std::string str_log;
			char str_buf[64] = { 0 };
			snprintf(str_buf, 63, " provider_type=%d, active_count=%d, alloc_type=%d, ",
				serviceConf.server_type, serviceConf.active_count, serviceConf.alloc_type);
			str_log += str_buf;
			for (int i = 0; i < serviceVect.size(); i++)
			{
				snprintf(str_buf, 63, " [idx=%d,svid=%d,isOk=%d],", i, serviceVect[i].svid, serviceVect[i].isOK);
				str_log += str_buf;
			}

			log_debug(" %s ", str_log.c_str());
#endif

			// send success
			m_servicePkgList.pop_front();
			delete pBuff;
		}
	}

	return 0;
}

int CNodeServer::PushPlayerMsg(unsigned int uid, CCPacketBase* pPkg)
{
	unsigned int seq = genSeq();

	NETOutputPacket pack;
	pack.Begin(ONLINE_STATE_QUERY_SYS, SERVER_TYPE_ONLINE, 0, m_type, TGlobal::_svid, seq);
	pack.WriteInt(uid);
	if (pPkg->packet_size() > 0)
	{
		pack.WriteBinary(pPkg->packet_buf(), pPkg->packet_size());
	}
	pack.End();


	return SendtoSrv(&pack, SERVER_TYPE_ONLINE, uid);

// 	CCSocketHandler* handler = NULL;
// 	BackTypeList::iterator itor = m_backTypeList.find(SERVER_TYPE_ONLINE);
// 	if (m_backTypeList.end() != itor)
// 	{
// 		if (itor->second.size() > 0)
// 		{
// 			int index = uid % itor->second.size();
// 			BackList::iterator itor1 = itor->second.begin();
// 			for (int i = 0; itor->second.end() != itor1 && i <= index; ++itor1, ++i)
// 			{
// 				if (index == i)
// 				{
// 					handler = itor1->second;
// 					break;
// 				}
// 			}
// 		}
// 	}
// 	else
// 	{
// 		SrvTypeList::iterator itor2 = m_svrTypeList.find(SERVER_TYPE_ONLINE);
// 		if (m_svrTypeList.end() != itor2)
// 		{
// 			if (itor2->second.size() > 0)
// 			{
// 				int index = uid % itor2->second.size();
// 				SrvDataList::iterator itor3 = itor2->second.begin();
// 				for (int i = 0; itor2->second.end() != itor3 && i <= index; ++itor3, ++i)
// 				{
// 					if (index == i)
// 					{
// 						handler = itor3->second->pHandler;
// 						break;
// 					}
// 				}
// 			}
// 		}
// 	}
// 	if (!handler)
// 		return -1;
// 
// 	handler->Send(&pack);
// 
// 	return 0;
}

int CNodeServer::BroadcastMessageDirect(CCPacketBase* pPkg, const unsigned short dst_type)
{
	BackTypeList::const_iterator itor = m_backTypeList.find(dst_type);
	if (m_backTypeList.end() != itor)
	{
		for (BackList::const_iterator itor1 = itor->second.begin(); itor->second.end() != itor1; ++itor1)
		{
			CClient* pBack = itor1->second;
			if (pBack)
				pBack->Send(pPkg);
		}
	}

	SrvTypeList::const_iterator itor2 = m_svrTypeList.find(dst_type);
	if (m_svrTypeList.end() != itor2)
	{
		for (SrvDataList::const_iterator itor3 = itor2->second.begin(); itor2->second.end() != itor3; ++itor3)
		{
			CClientHandler* handler = itor3->second->pHandler;
			if (handler)
				handler->Send(pPkg);
		}
	}

	return 0;
}

int CNodeServer::OnEventCallBack(stEventItem* pitem)
{
	if (!m_pHost)
		return -1;

	m_pHost->OnEventCallBack(pitem);
	return 0;
}

int CNodeServer::InitMtSend(void)
{
	m_pSignalEvent = new CSignalEvent();
	if (!m_pSignalEvent)
		return -1;

	int ret = m_pSignalEvent->create();
	if (0 != ret)
	{
		log_error("create signalevent fail");
		return -1;
	}
	
	m_pSignalEvent->m_pServer = this;

	if (CCReactor::Instance()->AttachPoller(m_pSignalEvent) < 0)
	{
		log_error("attach poller fail");
		return -1;
	}

	return 0;
}

unsigned int CNodeServer::genSeq(void)
{
	static unsigned int s_pushseq = 0;
	while (!(s_pushseq = ++s_pushseq % numeric_limits<unsigned int>::max()))
	{
	}

	return s_pushseq;
}